﻿
Imports System.Threading
Imports std

Public Class jobqueue
    Private Shared Function job_timeouted(ByVal job As indexedJob, ByVal ms As Int64) As Boolean
        assert(Not job Is Nothing)
        If Now.milliseconds() - job.ctorMs() >= ms Then
            raiseError(errorHandle.errorType.warning,
                       "job ", job.index(), " exceeded ", ms, " milliseconds limitation")
            Return True
        Else
            Return False
        End If
    End Function

    Private Shared Sub begin_eventComb(ByVal ms As Int64, ByVal d As void)
        assert(Not d Is Nothing)
        assert(ms > 0)
        assert_begin(New eventComb(Function() As Boolean
                                       Return waitfor(ms) AndAlso
                                              goto_next()
                                   End Function,
                                   Function() As Boolean
                                       d()
                                       Return goto_prev()
                                   End Function))
    End Sub

    Private Class pending_t
        Private Const maxQueueLength As Int64 = 1024
        Private Shared ReadOnly popPendingMs As Int64 = secondsToMilliseconds(10)
        Private Shared ReadOnly pendingMs As Int64 = secondsToMilliseconds(30)
        Private ReadOnly q As qless(Of indexedJob) = Nothing

        Public Sub New(ByVal name As String, ByVal lastPopMs As _do(Of Int64))
            assert(Not lastPopMs Is Nothing)
            q = New qless(Of indexedJob)()
            'planet is dead
            begin_eventComb(popPendingMs,
                            Sub()
                                If Now.milliseconds() - lastPopMs() >= popPendingMs Then
                                    If q.size() > maxQueueLength Then
                                        raiseError(errorHandle.errorType.warning,
                                                   "jobqueue.", name, ".pending has not been poped for ",
                                                   popPendingMs, " ms, clear.")
                                        Dim j As indexedJob = Nothing
                                        While q.size() > maxQueueLength AndAlso q.pop(j)
                                            j.cancel()
                                        End While
                                    End If
                                End If
                            End Sub)
            'too many jobs in the queue
            begin_eventComb(pendingMs,
                            Sub()
                                Dim j As indexedJob = Nothing
                                While q.pop(j)
                                    If job_timeouted(j, pendingMs) Then
                                        raiseError(errorHandle.errorType.warning,
                                                   "job in jobqueue.", name, ".pending is timeout already, cancel.")
                                        j.cancel()
                                    Else
                                        q.push(j)
                                        Exit While
                                    End If
                                End While
                            End Sub)
        End Sub

        Public Sub push(ByVal j As indexedJob)
            q.push(j)
        End Sub

        Public Function pop() As indexedJob
            Return q.pop()
        End Function
    End Class

    Private Class checkedOut_t
        Private Shared ReadOnly clearPendingMs As Int64 = secondsToMilliseconds(1)
        Private Shared ReadOnly checkedOutingMs As Int64 = minuteToMilliseconds(5)

        Private ReadOnly q As qless(Of indexedJob) = Nothing
        Private ReadOnly m As hashmapless(Of Int64, indexedJob, _1024, _int64_to_uint32) = Nothing

        Private Sub begin_eventComb(ByVal ms As Int64, ByVal d As _do(Of indexedJob, Boolean))
            assert(Not d Is Nothing)
            jobqueue.begin_eventComb(ms,
                                     Sub()
                                         Dim c As Int64 = 0
                                         c = q.size()
                                         Dim j As indexedJob = Nothing
                                         While q.pop(j) AndAlso c > 0
                                             If Not d(j) Then
                                                 q.push(j)
                                             End If
                                             c -= 1
                                         End While
                                     End Sub)
        End Sub

        Public Sub New(ByVal name As String)
            q = New qless(Of indexedJob)()
            m = New hashmapless(Of Int64, indexedJob, _1024, _int64_to_uint32)()
            'clear q
            begin_eventComb(clearPendingMs,
                            Function(ByRef j As indexedJob) As Boolean
                                Return j.finished() OrElse j.canceled()
                            End Function)
            'a job is taking too much time to be finished
            '5min, if a job need more time, it should not block
            begin_eventComb(checkedOutingMs,
                            Function(ByRef j As indexedJob) As Boolean
                                If j.canceled() OrElse j.finished() Then
                                    Return True
                                ElseIf job_timeouted(j, checkedOutingMs) Then
                                    raiseError(errorHandle.errorType.warning,
                                               "job in jobqueue.", name, ".checkedout is timeout already, cancel.")
                                    j.cancel()
                                    m.erase(j.index())
                                    Return True
                                Else
                                    Return False
                                End If
                            End Function)
        End Sub

        Public Sub insert(ByVal job As indexedJob)
            assert(Not job Is Nothing)
            m(job.index) = job
            q.push(job)
        End Sub

        Public Function [get](ByVal index As Int64) As indexedJob
            Return m(index)
        End Function

        Public Function [erase](ByVal index As Int64) As Boolean
            Return m.erase(index)
        End Function
    End Class

    Private ReadOnly pending As pending_t = Nothing
    Private ReadOnly checkedOut As checkedOut_t = Nothing
    Private index As Int64 = minInt64
    Private _trigger As void = Nothing
    Private lastPopMs As Int64 = 0

    Private Sub set_lastPopMs()
        lastPopMs = Now.milliseconds()
    End Sub

    Public Property trigger As void
        Set(ByVal value As void)
            assert((_trigger Is Nothing) <> (value Is Nothing))
            _trigger = value
        End Set
        Private Get
            Return _trigger
        End Get
    End Property

    Public Sub New(ByVal name As String)
        set_lastPopMs()
        pending = New pending_t(name, Function() lastPopMs)
        checkedOut = New checkedOut_t(name)
    End Sub

    Public Function nextJob(ByVal job As ijob) As indexedJob
        assert(Not job Is Nothing)
        Dim newIndex As Int64 = 0
        newIndex = Interlocked.Increment(index)
        Dim ij As indexedJob = Nothing
        ij = New indexedJob(newIndex, job)
        Return ij
    End Function

    Public Function push(ByVal job As indexedJob) As Boolean
        If job Is Nothing OrElse
           job.finished() OrElse job.canceled() OrElse
           job.job() Is Nothing OrElse Not job.job().valid() Then
            Return False
        End If

        pending.push(job)
        threadpoolRun(trigger)
        Return True
    End Function

    Public Function pop(ByRef job As indexedJob) As Boolean
        set_lastPopMs()
        Dim rtn As Boolean = False
        While Not rtn
            job = pending.pop()
            If job Is Nothing Then
                rtn = False
                Exit While
            Else
                assert(Not job.finished())
                rtn = Not job.canceled()
            End If
        End While
        If rtn Then
            assert(Not job Is Nothing)
            checkedOut.insert(job)
            job.checkOut()
        End If
        Return rtn
    End Function

    Public Function update(ByVal i As indexedJob, Optional ByVal d As void(Of indexedJob) = Nothing) As Boolean
        If i Is Nothing Then
            Return False
        Else
            Return update(i.index(), d)
        End If
    End Function

    Public Function update(ByVal index As Int64, Optional ByVal d As void(Of indexedJob) = Nothing) As Boolean
        Dim job As indexedJob = Nothing
        job = checkedOut.get(index)
        If Not job Is Nothing AndAlso checkedOut.erase(index) Then
            assert(Not job.finished() AndAlso Not job.canceled())
            void_(d, unref(job))
            'support to update a job but have not finish it yet.
            'since there may be some jobs have not been distributed to the right consumer.
            If Not (job.finished() OrElse job.canceled()) Then
                'parent job may cancel this job now
                push(job)
            End If

            Return True
        Else
            Return False
        End If
    End Function
End Class
